package test

import org.apache.spark.{SparkConf, SparkContext}

object demo6 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("RDDwork").setMaster("local")
    val sc = new SparkContext(conf)

    //    //将所有相同key的元素合并成一个元素
    //    //value会聚集到一个合集中
    //    var a1 = List(("zs", 98), ("ls", 60), ("zs", 45), ("ls", 78))
    //    var rdd1 = sc.parallelize(a1)  //将一个存在的集合，变成一个RDD
    //    var rdd2 = rdd1.groupByKey()  //无参数即可
    //    var rdd3 = rdd2.map(x=>{
    //      // x._1 //key  zs
    //      // x._2 //value(98, 45)
    //      //x._2.sum
    //      (x._1, x._2.sum)
    //    })
    //    rdd3.collect().foreach(println)


    //    //(1)union()将两个RDD合并成为一个新的RDD
    //    //    var rdd1 = sc.parallelize(Array(3, 4, 5))
    //    //    var rdd2 = sc.parallelize(Array(7, 8, 9))
    //    //    var rdd3 = rdd1.union(rdd2)
    //    //    rdd3.collect().foreach(println)
//        //（2）sortBy 将rdd中的元素按照某个规则进行排序 sortBy （排序函数 ，排序类型 ） 排序类型 --- 布尔型 升序 true 默认 降序 false
//        var a1 = Array(("zs", 34), ("ls", 88), ("ww", 66))
//        var rdd1 = sc.parallelize(a1)
//        var rdd2 = rdd1.sortBy(x => x._2, false)
//        rdd2.collect().foreach(println)
//        //（3）join算子相当于数据库中的内连接
//        var a1 = Array(("A", "a1"), ("B", "b1"), ("C", "c1"), ("D", "d1"), ("E", "e1"))
//        var a2 = Array(("A", "A1"), ("B", "B1"), ("C", "C1"), ("C", "C2"), ("C", "C3"), ("E", "E1"))
//        var rdd1 = sc.parallelize(a1)
//        var rdd2 = sc.parallelize(a2)
//        var rdd3 = rdd1.join(rdd2)
//        rdd3.collect().foreach(println)

//        //2、行动算子 （1 ）collect （） 执行计算 ，并将结果以数组的形式收集到 当前的drive中 （2 ）、reduce 对RDD中的元素进行聚合计算
//        var rdd1 = sc.parallelize(Array(2, 3, 4, 5, 6, 7, 8, 9))
//        //    var result=rdd1.reduce((x,y)=>x+y)
//        var result = rdd1.reduce(_ + _)
//        print(result)
//        //（3）、count 统计RDD集合中元素的数量
//        var rdd1 = sc.parallelize(Array(2, 3, 4, 5, 6, 7, 8, 9))
//        var num = rdd1.count()
//        print("长度：" + num)
//        //（4 ）、take （n）返回集合总前n组成的数据 -- 数组
//        var rdd1 = sc.parallelize(Array(2, 3, 4, 5, 6, 7, 8, 9))
//        var nums = rdd1.take(3)
//        nums.foreach(println)
        //(5) 、countByKey -- 返回map类型 按key值统计个数
        var a1 = List(("zs", 34), ("ls", 45), ("zs", 88))
        var rdd1 = sc.parallelize(a1)
        var result = rdd1.countByKey()
        for ((k, v) <- result) {
          println(k + "    " + v)
        }
  }
}
